package com.killer.mq;

import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.killer.enums.TaskStatus;
import com.killer.pojo.dto.BOrderDTO;
import com.killer.pojo.entity.Message;
import com.killer.service.MessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;


@Slf4j
@Service
public class MessageProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private MessageService messageService;

    public void asyncSendMQMessageSend(Long taskId, BOrderDTO orderDTO) {
        Message message = messageService.getById(taskId);
        //搜索条件
        message.setMessage_content(orderDTO.toString());
        //异步发送消息
        rocketMQTemplate.asyncSend(Constant.topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步消息发送成功:{}", sendResult);
                if(sendResult.getSendStatus() == SendStatus.SEND_OK){
                    updateMessageIsAcked(message);
                }else {
                    //更新重试次数
                    updateMessageRetryTimes(message);
                    new RuntimeException("消息发送失败");
                }
            }
            @Override
            public void onException(Throwable throwable) {
                log.info("异步消息发送失败:{}", throwable.getCause());
                //更新重试次数
                updateMessageRetryTimes(message);
                new RuntimeException("消息发送失败");
            }
        });
    }

    protected void updateMessageIsAcked(Message message){
        message.setIsAck(1);
        UpdateWrapper<Message> updateWrapper = new UpdateWrapper<>();
        updateWrapper.eq("id",message.getId());
        messageService.update(message,updateWrapper);
    }

    protected void updateMessageRetryTimes(Message message){
        message.setRetryTimes(message.getRetryTimes() + 1);
        UpdateWrapper<Message> updateWrapper = new UpdateWrapper<>();
        updateWrapper.eq("id",message.getId());
        messageService.update(message,updateWrapper);
    }


}
